package common.mq.basic.service.impl;

import java.util.GregorianCalendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;

import common.mq.basic.service.MessageService;
import common.mq.basic.service.QueueManager;
import common.mq.basic.util.QueueUtil;
import common.mq.master.service.IMasterService;
import common.mq.po.MessType;
import common.mq.po.Message;
import common.mq.po.Queue;
import common.mq.sender.service.ISenderService;

public class QueueManagerImpl implements QueueManager {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(QueueManagerImpl.class);
	
	private int poolSize = 10;
	
	@Autowired
	private IMasterService masterService;
	
	@Autowired
	private ISenderService senderService;
	
	@Autowired
	private MessageService messageService;

	@Override
	public Message popOneM(Queue q, String msgId) {
		if (q == null) {
			log.warn("[popOneM] q is null ");
			return null;
		}
		Message m = null;
		if (q.getWaitArea() != null
				&& q.getWaitArea().containsKey(msgId)) {
			m = q.getWaitArea().get(msgId);
			q.getWaitArea().remove(msgId);
		}
		log.debug("msg => " + m);
		return m;
	}

	@Override
	public void push(Message m, Queue q) throws Exception{
		if (m == null || m.getId() == null) {
			log.warn("[push] m is null " );
			throw new Exception("[push] m is null");
		}

		if (q.getWaitArea() != null
				&& q.getWaitArea().containsKey(m.getId())) {
			log.warn("[push] m had exist , m => {} ", m);
			throw new Exception("[push] m had exist , m => " + m);
		}
		
		if (q.getWaitArea() == null ){
			Map<String, Message> waitArea = new HashMap<String, Message>();
			q.setWaitArea(waitArea);
		}
		q.getWaitArea().put(m.getId(), m);
	}

	@Override
	public boolean checkSeqRelation(Queue q, Message m) {
		if(m.getType() != MessType.SEQ){
			return true;
		}
		Map<String, Message> areaMap = q.getWaitArea();
		if(areaMap == null){
			return true;
		}
		List<Message> mList = QueueUtil.getMessagesByCategory(areaMap, m.getCategory());
		if(mList == null || mList.isEmpty()){
			return true;
		}
		for(Message msg : mList){
			if(msg.getReceiveTime() < m.getReceiveTime()){
				return false;
			}
		}
		return true;
	}

	@Override
	public void push2Send(final String msgId,final  int qId,final  int sendTime) {
		log.debug("[push2Send] start , sendTime => "+sendTime+"s , time => " + new GregorianCalendar().getTime());
		//开启线程池
		ScheduledThreadPoolExecutor exec = new ScheduledThreadPoolExecutor(getPoolSize());
		/**
		 * 创建并执行在给定延迟后启用的一次性操作。
		 */
		exec.schedule(new Runnable() {
			public void run() {
				log.info("thread is runing , msgId=>{},qId=>{},time =>"+ new GregorianCalendar().getTime(), msgId, qId);
				try {
					int now = (int) new GregorianCalendar().getTimeInMillis() / 1000;
					// qId获取Queue
					Queue q = masterService.getQueue(qId);
					//队列关闭状态，则停止
					if(q.getStatus() == 2){
						return;
					}
					// 根据msgId 获取 Message
					Message m = popOneM(q, msgId);
					// 消息不存在或过期，则抛弃
					if (m == null || m.getOverdueTime() < now) {
						log.info("msg overdue , msg => " + m);
						return;
					}
					// 如果是时序消息，则需要检查它的时序
					if (m.getType() == MessType.SEQ && !checkSeqRelation(q, m)) {
						log.info("msg seq , msg => " + m);
						// 计算下次执行时间
						m = messageService.updateNextExecute(m,
								messageService.gainNextTime(m));
						// 存入等待区域
						push(m, q);
						push2Send(msgId, qId, m.getNextTime() - now);
						return;
					}
					if (!senderService.send(qId, m)) {
						// 计算下次执行时间
						m = messageService.updateNextExecute(m,messageService.gainNextTime(m));
						// 存入等待区域
						push(m, q);
						log.info("msg send fail , msg => " + m);
						push2Send(msgId, qId, m.getNextTime() - now);
						return;
					}
				} catch (Exception e) {
					log.warn("[push2Send] exception => ", e);
				}
			}
		}, sendTime, TimeUnit.SECONDS);
	}

	public void setPoolSize(int poolSize) {
		this.poolSize = poolSize;
	}

	public int getPoolSize() {
		return poolSize;
	}

	public void setMessageService(MessageService messageService) {
		this.messageService = messageService;
	}

	public MessageService getMessageService() {
		return messageService;
	}

	public void setMasterService(IMasterService masterService) {
		this.masterService = masterService;
	}

	public void setSenderService(ISenderService senderService) {
		this.senderService = senderService;
	}


}
